package plugin

import (
	"context"
	"encoding/gob"
	"net/rpc"
	"time"

	goplugin "github.com/hashicorp/go-plugin"
	"github.com/louloulin/dataflare/pkg/message"
)

// 注册gob类型
func init() {
	gob.Register(message.DataTypeJSON)
	gob.Register(message.DataTypeCSV)
	gob.Register(message.DataTypeXML)
	gob.Register(message.DataTypeBinary)
	gob.Register(map[string]interface{}{})
	gob.Register(map[string]string{})
	gob.Register([]interface{}{})
	gob.Register([]string{})
	gob.Register(PluginInfo{})
}

// HandshakeConfig 是用于identifying which plugin is which
const (
	PluginProtocolVersion = 1
)

// HandshakeConfig 是用于RPC握手的配置
// 它告诉插件系统是哪个插件以及应该使用什么版本
var HandshakeConfig = goplugin.HandshakeConfig{
	ProtocolVersion:  PluginProtocolVersion,
	MagicCookieKey:   "DATALINK_PLUGIN",
	MagicCookieValue: "datalink_plugin_v1",
}

// PluginMap 是可用插件的映射
var PluginMap = map[string]goplugin.Plugin{
	"processor": &ProcessorGRPCPlugin{},
	"connector": &ConnectorGRPCPlugin{},
}

// RPCProcessor 是处理器实现的接口
type RPCProcessor interface {
	Process(data interface{}) (interface{}, error)
}

// RPCConnector 是连接器实现的接口
type RPCConnector interface {
	Connect() error
	Disconnect() error
	Send(data interface{}) error
	Receive() (interface{}, error)
}

// RPCCommon 是所有插件共享的通用RPC接口
type RPCCommon interface {
	Initialize(config map[string]interface{}) error
	Start() error
	Stop() error
	GetInfo() (PluginInfo, error)
}

// ProcessorRPC 是客户端和服务器之间使用的RPC接口
type ProcessorRPC interface {
	RPCCommon
	Process(args interface{}, resp *interface{}) error
}

// ConnectorRPC 是客户端和服务器之间使用的RPC接口
type ConnectorRPC interface {
	RPCCommon
	Connect(args interface{}, resp *interface{}) error
	Disconnect(args interface{}, resp *interface{}) error
	Send(args interface{}, resp *interface{}) error
	Receive(args interface{}, resp *interface{}) error
}

// RPCClient 实现了processor/connector接口，并通过RPC通信
type RPCClient struct {
	client *rpc.Client
}

// Initialize 通过RPC调用实现Initialize方法
func (g *RPCClient) Initialize(ctx context.Context, config map[string]interface{}) error {
	var resp interface{}
	err := g.client.Call("Plugin.Initialize", config, &resp)
	return err
}

// Start 通过RPC调用实现Start方法
func (g *RPCClient) Start(ctx context.Context) error {
	var resp interface{}
	err := g.client.Call("Plugin.Start", new(interface{}), &resp)
	return err
}

// Stop 通过RPC调用实现Stop方法
func (g *RPCClient) Stop(ctx context.Context) error {
	var resp interface{}
	err := g.client.Call("Plugin.Stop", new(interface{}), &resp)
	return err
}

// GetInfo 通过RPC调用实现GetInfo方法
func (g *RPCClient) GetInfo() (PluginInfo, error) {
	var resp PluginInfo
	err := g.client.Call("Plugin.GetInfo", new(interface{}), &resp)
	return resp, err
}

// ProcessorRPCClient 实现处理器接口通过RPC通信
type ProcessorRPCClient struct {
	*RPCClient
}

// Process 通过RPC调用实现Process方法
func (g *ProcessorRPCClient) Process(ctx context.Context, input message.DataMessage) (message.DataMessage, error) {
	var resp map[string]interface{}
	// 将DataMessage转换为map，通过RPC传输
	args := map[string]interface{}{
		"id":        input.GetID(),
		"timestamp": input.GetTimestamp(),
		"type":      input.GetType(),
		"data":      input.GetData(),
		"metadata":  input.GetMetadata(),
	}

	err := g.client.Call("Plugin.Process", args, &resp)
	if err != nil {
		return nil, err
	}

	// 将返回的map转回DataMessage
	result := message.CreateBasicMessage()
	result.SetData(resp["data"])
	if ts, ok := resp["timestamp"].(int64); ok {
		result.CreatedAt = time.Unix(0, ts)
	}
	if metadata, ok := resp["metadata"].(map[string]interface{}); ok {
		result.SetMetadata(metadata)
	}
	return result, nil
}

// ConnectorRPCClient 实现连接器接口通过RPC通信
type ConnectorRPCClient struct {
	*RPCClient
}

// Connect 通过RPC调用实现Connect方法
func (g *ConnectorRPCClient) Connect(ctx context.Context) error {
	var resp interface{}
	err := g.client.Call("Plugin.Connect", new(interface{}), &resp)
	return err
}

// Disconnect 通过RPC调用实现Disconnect方法
func (g *ConnectorRPCClient) Disconnect(ctx context.Context) error {
	var resp interface{}
	err := g.client.Call("Plugin.Disconnect", new(interface{}), &resp)
	return err
}

// Send 通过RPC调用实现Send方法
func (g *ConnectorRPCClient) Send(ctx context.Context, data interface{}) error {
	var resp interface{}
	err := g.client.Call("Plugin.Send", data, &resp)
	return err
}

// Receive 通过RPC调用实现Receive方法
func (g *ConnectorRPCClient) Receive(ctx context.Context) (interface{}, error) {
	var resp interface{}
	err := g.client.Call("Plugin.Receive", new(interface{}), &resp)
	return resp, err
}

// RPCServer 是插件接口的RPC服务器实现
type RPCServer struct {
	Impl interface{}
}

// ProcessorRPCServer 是处理器插件的RPC服务器实现
type ProcessorRPCServer struct {
	Impl ProcessorPlugin
}

// Initialize 实现Initialize RPC方法
func (s *ProcessorRPCServer) Initialize(args map[string]interface{}, resp *interface{}) error {
	config := &PluginConfig{
		Config: args,
	}
	return s.Impl.Init(context.Background(), config)
}

// Start 实现Start RPC方法
func (s *ProcessorRPCServer) Start(args interface{}, resp *interface{}) error {
	return s.Impl.Start(context.Background())
}

// Stop 实现Stop RPC方法
func (s *ProcessorRPCServer) Stop(args interface{}, resp *interface{}) error {
	return s.Impl.Stop(context.Background())
}

// GetInfo 实现GetInfo RPC方法
func (s *ProcessorRPCServer) GetInfo(args interface{}, resp *PluginInfo) error {
	*resp = PluginInfo{
		ID:          s.Impl.ID(),
		Name:        s.Impl.Name(),
		Type:        s.Impl.Type(),
		Version:     s.Impl.Version(),
		Description: s.Impl.Description(),
		Config:      s.Impl.Config().Config,
	}
	return nil
}

// Process 实现Process RPC方法
func (s *ProcessorRPCServer) Process(args map[string]interface{}, resp *map[string]interface{}) error {
	// 将传入的map转换为DataMessage
	msg := message.CreateBasicMessage()
	msg.SetData(args["data"])
	if ts, ok := args["timestamp"].(int64); ok {
		msg.CreatedAt = time.Unix(0, ts)
	}
	if metadata, ok := args["metadata"].(map[string]interface{}); ok {
		msg.SetMetadata(metadata)
	}

	// 调用实现的Process方法
	result, err := s.Impl.Process(context.Background(), msg)
	if err != nil {
		return err
	}

	// 将返回的DataMessage转换为map
	*resp = map[string]interface{}{
		"id":        result.GetID(),
		"timestamp": result.GetTimestamp(),
		"type":      result.GetType(),
		"data":      result.GetData(),
		"metadata":  result.GetMetadata(),
	}
	return nil
}

// ConnectorRPCServer 是连接器插件的RPC服务器实现
type ConnectorRPCServer struct {
	Impl ConnectorPlugin
}

// Initialize 实现Initialize RPC方法
func (s *ConnectorRPCServer) Initialize(args map[string]interface{}, resp *interface{}) error {
	config := &PluginConfig{
		Config: args,
	}
	return s.Impl.Init(context.Background(), config)
}

// Start 实现Start RPC方法
func (s *ConnectorRPCServer) Start(args interface{}, resp *interface{}) error {
	return s.Impl.Start(context.Background())
}

// Stop 实现Stop RPC方法
func (s *ConnectorRPCServer) Stop(args interface{}, resp *interface{}) error {
	return s.Impl.Stop(context.Background())
}

// GetInfo 实现GetInfo RPC方法
func (s *ConnectorRPCServer) GetInfo(args interface{}, resp *PluginInfo) error {
	*resp = PluginInfo{
		ID:          s.Impl.ID(),
		Name:        s.Impl.Name(),
		Type:        s.Impl.Type(),
		Version:     s.Impl.Version(),
		Description: s.Impl.Description(),
		Config:      s.Impl.Config().Config,
	}
	return nil
}

// Connect 实现Connect RPC方法
func (s *ConnectorRPCServer) Connect(args interface{}, resp *interface{}) error {
	return s.Impl.Connect(context.Background())
}

// Disconnect 实现Disconnect RPC方法
func (s *ConnectorRPCServer) Disconnect(args interface{}, resp *interface{}) error {
	return s.Impl.Disconnect(context.Background())
}

// Send 实现Send RPC方法
func (s *ConnectorRPCServer) Send(args interface{}, resp *interface{}) error {
	return s.Impl.Send(context.Background(), args)
}

// Receive 实现Receive RPC方法
func (s *ConnectorRPCServer) Receive(args interface{}, resp *interface{}) error {
	result, err := s.Impl.Receive(context.Background())
	if err != nil {
		return err
	}
	*resp = result
	return nil
}

// ProcessorGRPCPlugin 是处理器插件的go-plugin实现
type ProcessorGRPCPlugin struct {
	goplugin.Plugin
	Impl ProcessorPlugin
}

// Server 返回一个可用于服务RPC调用的RPC服务器
func (p *ProcessorGRPCPlugin) Server(*goplugin.MuxBroker) (interface{}, error) {
	return &ProcessorRPCServer{Impl: p.Impl}, nil
}

// Client 返回一个与RPC服务器通信的接口实现
func (p *ProcessorGRPCPlugin) Client(b *goplugin.MuxBroker, c *rpc.Client) (interface{}, error) {
	return &ProcessorRPCClient{
		RPCClient: &RPCClient{client: c},
	}, nil
}

// ConnectorGRPCPlugin 是连接器插件的go-plugin实现
type ConnectorGRPCPlugin struct {
	goplugin.Plugin
	Impl ConnectorPlugin
}

// Server 返回一个可用于服务RPC调用的RPC服务器
func (p *ConnectorGRPCPlugin) Server(*goplugin.MuxBroker) (interface{}, error) {
	return &ConnectorRPCServer{Impl: p.Impl}, nil
}

// Client 返回一个与RPC服务器通信的接口实现
func (p *ConnectorGRPCPlugin) Client(b *goplugin.MuxBroker, c *rpc.Client) (interface{}, error) {
	return &ConnectorRPCClient{
		RPCClient: &RPCClient{client: c},
	}, nil
}
